package com.lez.coroutine02

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import org.junit.Test
import kotlin.system.measureTimeMillis

/**
 * 协程的上下文
 */
class CoroutineContextTest {

   @Test
   fun `test coroutine context`() = runBlocking<Unit> {
       //通过 + 来设置协程的上下文
      launch (Dispatchers.Default + CoroutineName("test")){
          println("i'm working in thread ${Thread.currentThread().name}")
      }
   }

   //Flow,emit发射器,返回多个值，并且是异步的，有值就返回
   suspend fun testFlow() = flow<Int>{
       for (i in 1..5){
           delay(1000)
           emit(i)
       }
   }
    //通过 collect收集flow发射的值，不再标有suspend
   @Test
   fun `test mutiple value`() = runBlocking<Unit>{
        launch {
            for (k in 1..6){
                delay(1000)
                println(k)
            }
        }
        testFlow().collect { it ->
           println(it)
        }
   }

   //流的连续性,从上游到下游每个过渡操作符都会处理每个发射的值，然后再交由末端操作符
    @Test
   fun `test flow continuation`()= runBlocking {
       (1..5).asFlow().filter {   //过渡操作符
           it % 2==0
       }.map {     //过渡操作符
           "string $it"
       }.collect {
           println("Collect $it")
       }
   }
   //流构建器
   @Test
   fun `test flow builder`() = runBlocking {
      flowOf("One","Two","Three","Four")
          .onEach { delay(1000) }
          .collect { println(it) }
   }

    //流的上下文,默认情况下 流的发射和流的收集都必须在同一个上下文，使用flowOn可以更改流发射的上下文
    fun `text flow context`() = flow<Int> {
        println("Flow started ${Thread.currentThread().name}")
            for (i in (1..5)) {
                delay(1000)
                emit(i)
            }
    }.flowOn(Dispatchers.IO)
    //测试流的上下文
    @Test
    fun `test flow context fun`() = runBlocking {
        println("Current Thread = ${Thread.currentThread().name}")
        `text flow context`().collect {
            println(it)
        }
    }

    //在指定协程中收集流  ,,定义一个事件源
    fun events() = (1..5)
        .asFlow()
        .onEach {
            delay(1000)
        }
        .flowOn(Dispatchers.Default)

    @Test
    fun `test flow launch`() = runBlocking {
       events().onEach { event -> println("Event $event ${Thread.currentThread().name}") }
           .launchIn(CoroutineScope(Dispatchers.Default))
           .join()
    }

    //背压
    fun flowTest1() = flow<Int>{
        for (i in 1..5){
            delay(100)
            emit(i)
            println("emit $i ${Thread.currentThread().name}")
        }
    }
    //测试背压
    @Test
    fun `test flow pressure`() = runBlocking {
       val time =  measureTimeMillis {
            flowTest1()
               // .buffer()  //先全部发射到管道 缓存起来，节省了发射的时间
               // .conflate() //合并处理最新值，中间值可能会丢失
                .collectLatest {  //取消并重新发射最后一个值
//                .collect {
                        value->
                    println("Collected $value ${Thread.currentThread().name}")
                    delay(1000)
                }
        }
       println("总执行时间：$time")
    }
}